package cn.baker.rabbitmq.listener;

import cn.baker.rabbitmq.entity.TransMessage;
import cn.baker.rabbitmq.service.TransMessageService;
import com.alibaba.fastjson.JSON;
import com.google.common.base.Throwables;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import javax.annotation.Resource;

/**
 * 监听消息消费者需要实现该抽象类
 * @author yuanyu
 */
@Slf4j
public abstract class AbstractMessageListener implements ChannelAwareMessageListener {

    @Resource
    private TransMessageService transMessageService;

    /**
     * 重新消费次数，超过后扔到死信队列
     */
    @Value("${trans.rabbitmq.consumer.reConsumerTimes:10}")
    private Integer reConsumerTimes;

    /**
     * 抽象方法
     *
     * @param message 消息
     */
    public abstract void receiveMessage(Message message);

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        MessageProperties messageProperties = message.getMessageProperties();
        long deliveryTag = messageProperties.getDeliveryTag();
        TransMessage transMessage =
                transMessageService.messageReceiveReady(
                        message.getMessageProperties().getMessageId(),
                        message.getMessageProperties().getReceivedExchange(),
                        message.getMessageProperties().getReceivedRoutingKey(),
                        message.getMessageProperties().getConsumerQueue(),
                        new String(message.getBody())
                );
        log.info("@mqLog 收到消息, 当前消息ID:[{}], 消费次数:[{}]", messageProperties.getMessageId(), transMessage.getSequence());
        try {
            receiveMessage(message);
            // 成功的回执
            channel.basicAck(deliveryTag, false);
            transMessageService.messageReceiveSuccess(transMessage.getId());
        } catch (Exception e) {
            log.error("@mqLog RabbitMQ消息消费失败, message:[{}], stackTrace:[{}]", e.getMessage(), Throwables.getStackTraceAsString(e));
            if (transMessage.getSequence() >= reConsumerTimes) {
                // 入死信队列
                channel.basicReject(deliveryTag, false);
                log.info("@mqLog 超过重新消费次数，丢弃消息message={}", JSON.toJSON(new String(message.getBody())));
            } else {
                // 重回到队列，重新消费，按照2的指数级递增
                Thread.sleep((long) (Math.pow(2, transMessage.getSequence()) * 1000));
                channel.basicNack(deliveryTag, false, true);
                log.info("@mqLog 消息重回到队列，message={}", JSON.toJSON(new String(message.getBody())));
            }
        }
    }
}
